#include "connecter.h"
#include "event2/event.h"
#include <unistd.h>
#include <thread>
#include "connect_handler.h"
#include "provider.h"
#include "spdlog/spdlog.h"
#include "deliver.h"
#include <iostream>
#include <sstream>

// 通知回调
static void NotifyCB(evutil_socket_t fd, short which, void *arg)
{
    Connecter *c = (Connecter *)arg;
    c->notify(fd, which);
}

Connecter::Connecter(int id, const std::string &amqp_address)
    : id_(id), amqp_address_(amqp_address), reconnect_interval_(1), seq_(0)
{
}

Connecter::~Connecter()
{
    if (channel_)
    {
        channel_->close();
        delete channel_;
    }
    if (connection_)
    {
        connection_->close();
        delete connection_;
    }
    if (handler_)
    {
        delete handler_;
    }
}

void Connecter::start()
{
    setup();
    connect();
    std::thread t(&Connecter::run, this);
    t.detach();
}

void Connecter::run()
{
    event_base_dispatch(base_);
    event_base_free(base_);
}

bool Connecter::setup()
{
    // 创建管道
    int fds[2];
    if (pipe(fds))
    {
        return false;
    }
    // 保存写入管道的文件描述符
    notify_send_fd_ = fds[1];
    // 创建libevent上下文，添加管道监听事件
    base_ = event_base_new();
    if (!base_)
    {
        return false;
    }
    event *ev = event_new(base_, fds[0], EV_READ | EV_PERSIST, NotifyCB, this);
    event_add(ev, 0);
    return true;
}

/**
 * Connecter类的notify方法，用于处理特定事件的通知。
 *
 * @param fd 文件描述符，表示发生事件的文件或套接字。
 * @param which 指定发生的事件类型，例如读或写事件。
 */
void Connecter::notify(int fd, short which)
{
    // 初始化一个缓冲区，用于读取数据
    char buf[1] = {0};
    // 从文件描述符fd中读取1字节的数据
    int re = read(fd, buf, 1);
    // 如果读取失败或没有数据，直接返回
    if (re <= 0)
    {
        return;
    }

    // 获取任务并初始化任务
    task_mutex_.lock();
    if (!delivers_.empty())
    {
        // 获取第一个任务
        Deliver *deliver = delivers_.front();
        // 从任务列表中移除该任务
        delivers_.erase(delivers_.begin());

        // 根据任务的标志决定执行的操作
        if (deliver->flag() == FLAG::OnlySend)
        {
            // 如果标志为1，表示只需发送数据
            send(deliver);
        }
        if (deliver->flag() == FLAG::SendAndReceived)
        {
            // 如果标志为2，表示需要发送数据并等待接收响应
            sendAndReceive(deliver);
        }
    }
    // 解锁任务互斥量，允许其他线程访问任务列表
    task_mutex_.unlock();
}

void Connecter::reconnect()
{
    reconnect_interval_ = (reconnect_interval_ * 2) % 30;
    std::this_thread::sleep_for(std::chrono::seconds(reconnect_interval_));
    if (channel_)
    {
        channel_->close();
        delete channel_;
    }

    if (connection_)
    {
        connection_->close();
        delete connection_;
    }

    if (handler_)
    {
        delete handler_;
    }
    connect();
    for (auto provider : providers_)
    {
        addProvider(provider);
    }
}

/**
 * 在Connecter中添加一个Provider
 *
 * 此函数负责将一个Provider对象添加到Connecter的providers列表中，如果该Provider不存在。
 * 它还会为Provider声明相应的交换机和队列，并绑定必要的回调函数来处理消息消费。
 *
 * @param provider 指向要添加的Provider对象的指针
 */
void Connecter::addProvider(Provider *provider)
{
    // 检查providers_列表中是否已存在相同的Provider
    bool is_exist = false;
    for (int i = 0; i < providers_.size(); i++)
    {
        // 如果发现应用名和子名都相同的Provider，则标记为已存在
        if (providers_[i]->appName() == provider->appName() && providers_[i]->subName() == provider->subName())
        {
            is_exist = true;
        }
    }

    // 如果Provider不存在于列表中，则将其添加到providers_列表中
    if (!is_exist)
    {
        providers_.push_back(provider);
    }

    // 声明一个直接类型的交换机，名为provider的appName
    channel_->declareExchange(provider->appName(), AMQP::direct).onSuccess([provider]()
                                                                           { spdlog::info("{} exchange declare success", provider->appName()); });

    // 声明一个队列，名为provider的appName加subName，并设置自动删除属性
    channel_->declareQueue(provider->appName() + "." + provider->subName(), AMQP::autodelete)
        .onSuccess([](const std::string &name, uint32_t messagecount, uint32_t consumercount)
                   { spdlog::info("{} queue declare success", name); });

    // 将队列绑定到交换机上，使用provider的subName作为路由键
    channel_->bindQueue(provider->appName(), provider->appName() + "." + provider->subName(), provider->subName())
        .onSuccess([this]()
                   {
            // 重连间隔重置
            reconnect_interval_ = 1; });

    // 开始消费队列中的消息，不需确认消息消费
    channel_->consume(provider->appName() + "." + provider->subName(), AMQP::noack)
        .onReceived([provider, this](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered)
                    {
                        // 解析接收到的消息体为ProviderData对象
                        ProviderData req;
                        req.data = message.body();
                        req.size = message.bodySize();

                        // 用于存储Provider处理后的响应数据
                        ProviderData res;

                        // 返回true表示需要应答，则创建一个响应的Envelope并尝试回复
                        if(provider->received(&req, &res))
                        {
                            AMQP::Envelope envelope(res.data, res.size);

                            // 如果原始消息有相关ID，则设置回复消息的相关ID
                            if(message.hasCorrelationID())
                            {
                                envelope.setCorrelationID(message.correlationID());
                            }

                            // 如果原始消息有回复地址，则发布回复消息到该地址
                            if(message.hasReplyTo()) 
                            {
                                channel_->publish("", message.replyTo(), envelope);
                            }
                        } });
}

/**
 * 添加一个投递对象到连接器中
 *
 * 此函数用于将一个投递对象（Deliver）添加到连接器的投递对象列表中
 * 它通过使用互斥锁来保护投递对象列表，以确保线程安全
 * 在添加完成后，通过写入通知管道来唤醒等待的线程
 *
 * @param deliver 指向要添加的投递对象的指针
 */
void Connecter::addDeliver(Deliver *deliver)
{
    task_mutex_.lock();

    delivers_.push_back(deliver);

    task_mutex_.unlock();

    // 通过写入通知管道来唤醒等待的线程
    write(notify_send_fd_, "a", 1);
}

void Connecter::connect()
{
    handler_ = new ConnectHandler(base_, this);
    connection_ = new AMQP::TcpConnection(handler_, AMQP::Address(amqp_address_));
    channel_ = new AMQP::TcpChannel(connection_);
}

/**
 * @brief 发送消息到指定的应用程序和订阅者
 *
 * 本函数通过AMQP协议，将消息发送到指定的应用程序和订阅者。它首先检查是否有有效的通信信道，
 * 然后开始一个事务，创建一个包含消息数据的信封，并通过信道发布该信封。发布操作是异步的，
 * 当发布成功后，它会通过回调函数通知，此时会设置消息为已发送状态，并通过条件变量通知等待线程。
 *
 * @param deliver 消息对象，包含要发送的数据、应用程序名称、订阅者名称等信息
 */
void Connecter::send(Deliver *deliver)
{
    // 检查是否有有效的通信信道
    if (channel_)
    {
        // 开始一个新的事务，以确保消息发送的原子性
        channel_->startTransaction();

        AMQP::Envelope envelope(deliver->data(), deliver->size());

        channel_->publish(deliver->appName(), deliver->subName(), envelope);

        // 当消息发布成功后，提交事务，并在成功回调中设置消息为已发送状态
        channel_->commitTransaction().onSuccess(
            [deliver]()
            { 
                // 加锁以保护共享资源，设置消息为已发送状态，并通过条件变量通知等待线程
                std::lock_guard<std::mutex> lock(*deliver->mutex());
                deliver->setReplied(true);
                deliver->cond()->notify_one(); });
    }
}

/**
 * @brief 发送消息并接收回复
 *
 * 该函数通过RabbitMQ通道发送消息并监听回复。它首先检查通道是否存在，
 * 然后声明一个排他且自动删除的队列。成功声明队列后，它将消息封装到
 * 一个信封中，设置相关ID和回复队列名称，然后发布消息。同时，它开始
 * 监听回复队列，接收到回复后，将回复数据存储到deliver对象中，并通知
 * 等待线程，最后删除队列。
 *
 * @param deliver 消息传递对象，包含消息数据、大小及回复相关的信息
 */
void Connecter::sendAndReceive(Deliver *deliver)
{
    // 检查通道是否存在
    if (!channel_)
    {
        return;
    }

    // 声明一个排他且自动删除的队列
    channel_->declareQueue(AMQP::exclusive | AMQP::autodelete)
        .onSuccess([this, deliver](const std::string &queue_name, uint32_t message_count, uint32_t consumer_count)
                   {
            // 将deliver中的消息数据封装到信封中
            AMQP::Envelope envelope(deliver->data(), deliver->size());
            // 生成序列号
            int seq = seq_.fetch_add(1, std::memory_order_relaxed) + 1;
            // 设置信封的关联ID和回复队列名称
            envelope.setCorrelationID(std::to_string(seq));
            envelope.setReplyTo(queue_name);
            // 发布消息
            channel_->publish(deliver->appName(), deliver->subName(), envelope);

            channel_->consume(queue_name, AMQP::noack).onReceived([deliver, this, queue_name](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered){
                deliver->reply()->data = message.body();
                deliver->reply()->size = message.bodySize();
                // 锁定互斥锁，设置回复标志并通知等待线程
                std::lock_guard<std::mutex> lock(*deliver->mutex());
                deliver->setReplied(true);
                deliver->cond()->notify_one();
                // 删除队列
                channel_->removeQueue(queue_name);
            }); });
}
